package com;

import com.alibaba.fastjson2.JSON;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.Test;

import java.util.Properties;
import java.util.concurrent.*;

public class CustomProducer {
    public static void main(String[] args) {
//        Order order = new Order(1, 11, 5, 200.00);
//        System.out.println(JSON.toJSONString(order));

//        new Thread(new Runnable() {
//            @Override
//            public void run() {
//                System.out.println("aaaa");
//            }
//        }).start();

        // 针对线程提供了更加丰富的功能，
        FutureTask<String> futureTask =
                new FutureTask<>(new Callable<String>() {
                    public String call() {
                        System.out.println("abcd");
                        // System.out.println(1/0);
//                        try {
//                            Thread.sleep(1000);
//                        } catch (InterruptedException e) {
//                            e.printStackTrace();
//                        }

                        return "success";
                    }
                });

        // 使用 futureTask 来启动线程
        Thread thread = new Thread(futureTask);
        thread.start();

        // get获取执行的结果，main线程调用get方法，那么就会等待thread线程执行完毕，接收结果
        // 可以捕获到子线程抛出的异常
        try {
            System.out.println("isDone:" + futureTask.isDone());
            String result = futureTask.get(200, TimeUnit.MILLISECONDS);
            System.out.println("result:" + result);
        } catch (InterruptedException e) {
            System.out.println("捕获异常 InterruptedException");
            e.printStackTrace();
        } catch (ExecutionException e) {
            System.out.println("捕获异常 ExecutionException");
            System.out.println("记录异常信息到log，");
            e.printStackTrace();
        } catch (TimeoutException e) {
            System.out.println("捕获异常 TimeoutException");
            System.out.println("记录异常信息到log，");
            e.printStackTrace();
        }


    }

    @Test
    public void test2() {
        System.out.println(StringSerializer.class.getName());
    }


    @Test
    public void test1() {
        String topic = "order";

        // 创建properties，在props对象中可以自定义的修改kafka提供的属性默认值
        Properties props = new Properties();

        // 配置参数
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node01:9092,node02:9092,node03:9092");

        // 设置 发出消息持久化机制参数
        props.put(ProducerConfig.ACKS_CONFIG, "0");

        // 发送重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        // 重试的间隔时间
        props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 200);

        // buffer缓冲区大小
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

        // 指定key.serializer，value.serializer
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        new Thread().setDaemon(true);
        int numMsg = 20;

        // 异步的逻辑需要处理几次
        CountDownLatch countDownLatch = new CountDownLatch(numMsg);
        for (int i = 1; i <= numMsg; i++) {
            // 发送的消息，是一条订单记录
            Order order = new Order(i, 10 + i, 5 + i, 200.00 * i);

            // 需要指定key和value的类型, KafkaProducer指定了类型后，那么recored的key、value类型也就确定了
            KafkaProducer<String, String> producer = new KafkaProducer<>(props);

            // 创建消息对象，KafkaRecord
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, order.getOrderId() + "",
                    JSON.toJSONString(order));


            // 异步发送,  后台线程,坦克大战
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if(exception != null) {
                        exception.printStackTrace();
                    } else {
                        System.out.println("topic:" + metadata.topic() + "---partition:" + metadata.partition() +
                                "---offset:" + metadata.offset());
                        // 每次异步逻辑处理完毕
                        countDownLatch.countDown();
                    }
                }
            });




            // 同步发送消息,程序阻塞等待发送的结果，后续代码要基于发送成功or失败，来确定后续代码
//            Future<RecordMetadata> future = producer.send(record);
//
//            // 等待消息发送的结果
//            try {
//                // 返回消息的元数据,get没有异常，那么就说明消息发送成功
//                RecordMetadata metadata = future.get();
//                System.out.println("topic:" + metadata.topic() + "---partition:" + metadata.partition() +
//                        "---offset:" + metadata.offset());
//                // 成功的业务逻辑
//
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//                // 发送失败，记录信息到log
//                System.out.println(e.getMessage());
//            } catch (ExecutionException e) {
//                e.printStackTrace();
//            }
        }

//        try {
//            Thread.sleep(500);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }

        try {
            // countDownLatch.await();
            // 设置超时时间
            countDownLatch.await(2, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("main 线程之后的逻辑代码！！！");


    }

}

